package org.example
import org.apache.spark.sql.SparkSession


import java.util.regex.Pattern

object data1_movies {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder
      .master("local[*]")
      .appName("spark")
      .getOrCreate()

    val sc = spark.sparkContext

//
    val  filePath="src/main/resources/"
    val moviesRDD=sc.textFile(filePath+"movies.dat")
    val occupationsRDD=sc.textFile(filePath+"occupations.dat")
    val ratingsRDD=sc.textFile(filePath+"ratings.dat")
    val usersRDD=sc.textFile(filePath+"users.dat")
    println("电影名称:"+moviesRDD.count()+"\n"+"用户总数"+usersRDD.count())
    println("评分总数："+ratingsRDD.count()+"\n"+"职业数"+occupationsRDD.count())
  //分析 某用户评分以及对应电影详情（id，电影名，类型
//    val num=ratingsRDD.map(_.split("::"))
//      .map(user =>( user(1),user(0)))//转换key value型 电影id 用户 id
//       .filter(_._2.equals("3088")) //用户id为评分总数
//      println("用户id为评分总数"+num.count())
 //用户3088评分电影的详情 （电影名 类型）
//    val moviesInfo=moviesRDD.map(_.split("::"))
//      .map(movies=>(movies(0),(movies(1),movies(2))))//moviesId,(userId,(Title,Drama)
//    val res=num.join(moviesInfo)
//      .map(item=>(item._1,item._2._1,item._2._2._1,item._2._2._2))
//    println("电影详情\n")
//    res.take(3).foreach(println)
//分析三 找出评分最高的前10名电影，观影人数最多的前10名的电影
//val rating = ratingsRDD.map(x => x.split("::")).map { x => {
//  (x(0), x(1), x(2)) //(UserID,MovieID,Rating) } }. cache()
//  println("平均得分最高的前10名的电影名称简单版")
//  //(MovieId,( Rating, 1))(1200，(4.0,1))
//  rating.map(x => (x._2, (x._3.toDouble, 1))) //
//  // ( MovieId,( 总分, 总次数) )
//  .reduceByKey((x, y) => {  (x._1 + y._1, x._2 + y._2)
//  }) // (  平均分, MovieId )
//  .map(x => (x._2._1 / x._2._2, x._1))
//    .sortByKey(false)
//    .take(5)
//    .foreach
//  println("按平均分取前10部电影输出详情：(平均分，(movieId,Title,Genres,总分，总次 数))")
//
//  // MovieID: ： Title: ： Genres
//  val moviesInfo = moviesRDD.map(x => x.split("::"))
//    .map(x => {
//      (x(0), (x(1), x(2)))
//    })
//  val ratingsInfo = rating.map(x => (x._2, (x._3.toDouble, 1))) //( MovieId,( Rating, 1))(1252，(4.0,1))
//    .reduceByKey((x, y) => {
//      (x._1 + y._1, x._2 + y._2)
//    }) //(MovieId,(总分，总次数))
//    .map(x => (x._1, (x._2._1 / x._2._2, x._2._1, x._2._2))) //(MovieId, (平均分，总分，总次数))
//  moviesInfo.join(ratingsInfo).map(info => {
//    (info._2._2._1, (info._1, info._2._1._1, info._2._1._2, info._2._2._2, info._2._2._3)) //(平均分,(movieId,Title,Genres,总分,总次数)) })
//      .sortByKey(false)
//      .take(5)
//      .foreach(println)
//    println("观影人数最多的前 10 部电影") //(UserID,MovieID,Rating)
//
//    val watchViewsInfo = rating.map(x => {
//        (x._2, 1)
//      }).reduceByKey((x, y) =>× +y) //(MovieId,总次数)
//      .map(x => (x._2, x._1))
//      .sortByKey(false)
//      .take(5) //5名
//    watchViewsInfo.foreach(println(_))
//    println("===================>")
//
//
//    rating.map(x => (x._2, 1))
//    .reduceByKey((x, y) => {
//      (x + y)
//    })
//      .sortByKey(false)
//      .take(10)
//      .foreach(println) // 286->999
//    println("详情的输出(观影人数,电影编号)")
//    //输出（总次数，(MovieID,title,Genres,总分，平均分）)
//    // ratingsInfo(MovieId,(平均分，总分，总次数))
//    // (MovieID,((Title,Genres),(平均分，总分，总次数)))
//    // ratingsInfo （ MovieId,(平均分，总分，总次数）)
//    // moviesInfo MovieID: ：Title: ： Genres
//
//    moviesInfo.join(ratingsInfo).map(x => {
//        (x._2._2._3, (x._2._1._1, x._2._1._2, x._2._2._1,x._2._2._2))
//      }).sortByKey(false)
//      .take(10)
//      .foreach(println)
    // 统计不同电影类型数量
    val moviesStyle=moviesRDD.map(_.split("::"))
  .map(tp=>(tp(0),tp(2))) //只留编号类型
  .flatMapValues(sp=>sp.split("\\|"))//切分电影类型
  .map(p=>(p._2,1))//转换键值对
  .reduceByKey((x,y)=>x+y)//聚合
  .filter(_._1.equals("Comedy"))
    moviesStyle.take(3).foreach(println)
    //打印喜剧篇（comdey）的数量
    val pattern=Pattern.compile("(.*)(\\(\\d{4}\\))")
    val movieYear=moviesRDD.map(_.split("::"))
      .map(x=>(x(1),1))
      .map(x=>{
        var name=""
        var year=""
        val matcher=pattern.matcher(x._1)
        if (matcher.find()){
          name=matcher.group(1)
          year=matcher.group(2)
          year=year.substring(1,year.length-1)

        }
        if (year==""){
          (-1,1)
        }else{
          (year.toInt,1)
        }
      })
      .reduceByKey((x,y)=>x+y)
      .filter(_._1==2000)
      .map(_._2)
     // .sortByKey()

   movieYear.foreach(println)
    sc.stop()
  }
}




